package com.wyz.socket.core.impl;

import com.wyz.socket.core.api.*;
import com.wyz.socket.core.base.*;
import com.wyz.socket.core.base.exceptions.EmptyIoArgsException;
import com.wyz.socket.core.utils.Log;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 通道适配器
 */
public class SocketChannelAdapter implements Sender, Receiver {

    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final IoProvider ioProvider;

    private final SocketChannel channel;

    private final List<Close> dispatcher;

    private final AbstractProviderCallback receiveCallback;
    private final AbstractProviderCallback sendCallback;

    public SocketChannelAdapter(SocketChannel channel, IoProvider ioProvider, Close close) {
        this.ioProvider = ioProvider;
        this.channel = channel;
        this.dispatcher = new ArrayList<>();
        this.dispatcher.add(close);
        this.receiveCallback = new ReceiveProviderCallback(ioProvider, channel);
        this.sendCallback = new SendProviderCallback(ioProvider, channel);
    }

    @Override
    public void registerSocketListener(Close close) {
        this.dispatcher.add(close);
    }

    @Override
    public void unregisterSocketListener(Close close) {
        this.dispatcher.remove(close);
    }

    @Override
    public void setReceiveListener(IoArgs.IoBufferEventProcessor processor) {
        this.receiveCallback.eventProcessor = processor;
    }

    @Override
    public void setSendListener(IoArgs.IoBufferEventProcessor processor) {
        this.sendCallback.eventProcessor = processor;
    }

    @Override
    public void postSendAsync() throws Exception {
        if (isClosed.get() || !channel.isOpen()) {
            throw new IOException("Current channel is closed!");
        }
        sendCallback.checkAttachSuccess();
        //sendCallback.run();
        ioProvider.register(sendCallback);
    }


    @Override
    public void postReceiveAsync() throws Exception {
        if (isClosed.get() || !channel.isOpen()) {
            throw new IOException("Current channel is closed!");
        }
        receiveCallback.checkAttachSuccess();
        ioProvider.register(receiveCallback);
    }

    @Override
    public void stopAsync() throws IOException {
        if (isClosed.get()) {
            throw new IOException("Current channel is closed!");
        }
        ioProvider.unregister(channel);
    }

    @Override
    public void close() {
        if (isClosed.compareAndSet(false, true)) {
            ioProvider.unregister(channel);
            this.receiveCallback.eventProcessor = null;
            this.sendCallback.eventProcessor = null;
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            for (Close close : dispatcher) {
                close.onClose(channel);
            }
            dispatcher.clear();
        }
    }

    abstract class AbstractProviderCallback extends ProviderCallback {
        volatile IoArgs.IoBufferEventProcessor eventProcessor;

        public AbstractProviderCallback(IoProvider provider, SocketChannel channel, int ops) {
            super(provider, channel, ops);
        }

        @Override
        protected final boolean onRun(IoArgs args) {
            if (isClosed.get()) {
                return false;
            }
            IoArgs.IoBufferEventProcessor processor = this.eventProcessor;
            if (processor == null) {
                return false;
            }
            processor.onstart();
            if (args == null) {
                args = processor.providerIoArgs();
            }
            try {
                if (args == null) {
                    throw new EmptyIoArgsException("read from socket but IoArgs is null ");
                }
                int consumeChannelSize = consumeChannel(args, channel);

                if (args.hasRemaining() && (consumeChannelSize == 0 || args.isNeedConsumeRemaining())) {
                    this.args = args;
                    return true;
                } else {
                    return processor.onConsumeCompleted(args);
                }

            } catch (Throwable e) {
                Log.println("SocketChannelAdapter" , "onRun" , e.getClass().getSimpleName()+"  "+e.getMessage());
                if (processor.onConsumeFailed(e)) {
                    SocketChannelAdapter.this.close();
                }
                return false;
            } finally {
                processor.onEnd();
            }
        }

        protected abstract int consumeChannel(IoArgs args, SocketChannel channel) throws IOException;

        @Override
        public void onThrowException(Throwable throwable) {
            IoArgs.IoBufferEventProcessor eventProcessor = this.eventProcessor;
            if (eventProcessor != null) {
                boolean b = eventProcessor.onConsumeFailed(throwable);
                if (b) {
                    SocketChannelAdapter.this.close();
                }
            }
        }
    }

    class ReceiveProviderCallback extends AbstractProviderCallback {

        public ReceiveProviderCallback(IoProvider provider, SocketChannel channel) {
            super(provider, channel, SelectionKey.OP_READ);
        }

        @Override
        protected int consumeChannel(IoArgs args, SocketChannel channel) throws IOException {
            return args.readFrom(channel);
        }
    }

    class SendProviderCallback extends AbstractProviderCallback {

        public SendProviderCallback(IoProvider provider, SocketChannel channel) {
            super(provider, channel, SelectionKey.OP_WRITE);
        }

        @Override
        protected int consumeChannel(IoArgs args, SocketChannel channel) throws IOException {
            return args.writeTo(channel);
        }
    }
}
